package com.practice.gmall.realtime.joinAPI;

import com.practice.gmall.realtime.app.BaseSQLApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * ClassName: Tutorial_InnerJoinToKafka
 * Package: com.practice.gmall.realtime.joinAPI
 * Description: write the result from inner join to Kafka
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-14 9:26
 */
public class Tutorial_InnerJoinToKafka extends BaseSQLApp {

    public static void main(String[] args) {
        new Tutorial_InnerJoinToKafka().init(3337, "Tutorial_InnerJoinToKafka", 2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15));

        tEnv.executeSql("create table left_tb( " +
                "    id string, " +
                "    name string " +
                ")with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = 'left_tb', " +
                "    'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "    'properties.group.id' = 'Tutorial_InnerJoin', " +
                "    'scan.startup.mode' = 'latest-offset', " +
                "    'format' = 'csv' " +
                ")");

        tEnv.executeSql("create table right_tb( " +
                "    id string, " +
                "    gender string " +
                ")with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = 'right_tb', " +
                "    'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "    'properties.group.id' = 'Tutorial_InnerJoin', " +
                "    'scan.startup.mode' = 'latest-offset', " +
                "    'format' = 'csv' " +
                ")");

        tEnv.executeSql("create table inner_join_result(  " +
                "    id string,  " +
                "    name string,  " +
                "    gender string  " +
                ")with (  " +
                "    'connector' = 'kafka',  " +
                "    'topic' = 'inner_join_result',  " +
                "    'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                "    'format' = 'json'  " +
                ")");

        Table result = tEnv.sqlQuery("select l.id, name, gender from left_tb l inner join right_tb  r on l.id = r.id");

        result.executeInsert("inner_join_result");
    }
}
